package com.aotain.jupiter.flink;

import java.util.HashSet;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearcher;
import com.aotain.common.CommonFunction;
import com.aotain.jupiter.common.PostTuple;

/**
 * 
     *@Title:
     *@Description:
     *@Author:Turk
     *@Since:2017年7月25日
     *@Version:1.1.0
 */
public class SourceDataPost implements MapFunction<String, Tuple2<String,PostTuple>> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6137189732292460920L;


//	static HashSet<String> IPSet;
//	static HashSet<String> WhiteIPSet;
	final long timeInterval = 1000*15;
	String filterIpSql ;
	String whiteIpSql;
	String dbJson;
	public SourceDataPost(String dbjson) {	
		dbJson = dbjson;
	}

	@Override
	public Tuple2<String,PostTuple> map(String value) throws Exception {
		// TODO Auto-generated method stub

		String refer = "";             String sourceAreaCountry = null;    String domainname = "";                    
		String dip = "";               String sourceAreaCity = null;       String url = "";                           
		String sip = "";               String sourceAreaName = null;       String type = "";                          
		String userAgent = "";         String destAreaCountry = null;      String createtime = "";         
		String cookie = "";            String destAreaCity = null;         String idcno = "";              
		String dipL = "";              String destAreaName = null;         String post = "";	          
		String sipL = "";              String gisstr = null;               String dsip = "";               
		String dport = "";             String gisstrBase64 = null;         String areaName_d = null;       
		String sport = "";             String Sourcelon = null;            long upStreamOctets = 0;        
		int status ;                   long upStreamPacket = 0;            String areaName_s = null;           
		String Sourcelat = null;       String sourceAreaId = null;         String destAreaId = null;          
		String Destlon = null;         String sourceAreaProvinceId = null; String destAreaProvinceId = null;  
		String Destlat = null;         String sourceAreaCityId = null;     String destAreaCityId = null;      
		String delaytime = null;
		Tuple2<String,PostTuple> ret = new Tuple2<String,PostTuple>();

		/*POST-DATA
		 * 0 username
		   1 prototype
		   2 srcip
		   3 destip
		   4 srcport
		   5 destport
		   6 host
		   7 url
		   8 refer
		   9 useragent
		   10 cookie
		   11 createtime
		   12 postlen
		   13 postcont
		 * */
		String[] splits = value.split("\\|",-1);
		//		System.out.println("***value" + value);

		String username = splits[0];
//			String ipRex = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";   
//			Pattern pattern = Pattern.compile(ipRex);   
//			Matcher matcher = pattern.matcher(username); 
		String prototype = splits[1];
		if(prototype.equals("6") && splits.length >= 14) {//post
			//if(splits.length ==43) {//http
			dip = splits[3];
			dport = splits[5];
			sip = splits[2];
			sport = splits[4];
			url = splits[7];
			createtime= splits[11];
			idcno = "446605";
			userAgent = splits[9];
			refer = splits[8];
			cookie = splits[10];
			post = splits[13];
		} else {
			return ret;
		}
			
			
		dsip = dip+"#"+sip;
		
		long currenttime = System.currentTimeMillis();
		
			
		//if(!IPFilterSingleUtil.getInstance(filterIpSql,whiteIpSql, dbJson).WhiteIpSet.contains(dsip)) {
			if(!"".equals(dport) && !"".equals(sport) ) {
				IPSearcher ipsearcher = IPSearcher.getInstance(dbJson);
				IPDatabase SourceArea = ipsearcher.SearchIPDB(sip);
				IPDatabase DestArea = ipsearcher.SearchIPDB(dip);
//				HashSet<String> IPSet = IPFilterSingleUtil.getInstance(filterIpSql,whiteIpSql, dbJson).IPSet;
//				if((IPSet == null || IPSet.contains(dip))) {

					if(SourceArea != null) {
						sourceAreaCountry = SourceArea.getCountryName();
						sourceAreaCity = SourceArea.getCityName();
						//						sourceAreaProvince = SourceArea.getProviceName();
						if(sourceAreaCity != null) {
							sourceAreaName  = sourceAreaCity;
							sourceAreaId = SourceArea.getCityID()+"";
						} else {
							sourceAreaName = sourceAreaCountry;
							sourceAreaId = SourceArea.getCountryID()+"";
						}
						sourceAreaCityId = sourceAreaId;
						Sourcelon = String.valueOf(SourceArea.getLon());//经度
						Sourcelat = String.valueOf(SourceArea.getLat());//纬度

						if(!"".equals(SourceArea.getProviceID()) && SourceArea.getProviceID() !=0 ) {
							sourceAreaId = SourceArea.getProviceID()+"";
							areaName_s = SourceArea.getProviceName().replace("省", "");
							areaName_s = areaName_s.replace("市", "");
						} else {
							areaName_s = SourceArea.getCountryName();
							sourceAreaId = SourceArea.getCountryID()+"";
						}
						sourceAreaProvinceId = sourceAreaId;
					}   

					if(DestArea != null ) {    
						destAreaCountry = DestArea.getCountryName();
						destAreaCity = DestArea.getCityName();
						if(destAreaCity != null) {
							destAreaName = destAreaCity;
						} else {   
							destAreaName = destAreaCountry;
						}
						Destlon = String.valueOf(DestArea.getLon());
						Destlat = String.valueOf(DestArea.getLat());
					}  

					if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null && areaName_s != null) {
						gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
						gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
						
						PostTuple tuple = new PostTuple();
						tuple.setDestIP(dip);
						tuple.setSourceIP(sip);
						tuple.setDestPort(dport);
						tuple.setSourcePort(sport);
						tuple.setGIS(gisstrBase64);
						tuple.setUrl(url);
						tuple.setIDC(idcno);
						tuple.setRefer(refer);
						tuple.setUserAgent(userAgent);
						tuple.setCookie(cookie);
						tuple.setPost(post);
						
						ret.setFields(dip, tuple);
					}
//				}
//			}
			
		}
		return ret;
	}

}
